/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl.statistics;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import spring.cloud.tasks.api.database.entity.StatisticsDto;
import spring.cloud.tasks.api.database.service.StatisticsDtoService;
import spring.cloud.tasks.api.domain.*;
import spring.cloud.tasks.api.service.helper.DashboardServiceHelper;
import spring.cloud.tasks.api.service.impl.statistics.analyzer.StatisticsModel;
import spring.cloud.tasks.api.utils.StatisticsTableKeyConstant;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author timmy.hu
 */
public class StatisticsPersistence {

    private static final Logger log = LoggerFactory.getLogger(StatisticsPersistence.class);

    @Resource
    private StatisticsDtoService statisticsService;

    public void persist(StatisticsModel statisticsModel, ZooKeeperCluster zkCluster) {
        List<JobStatistics> jobList = statisticsModel.getJobStatisticsAnalyzer().getJobList();
        List<ExecutorStatistics> executorList = statisticsModel.getExecutorInfoAnalyzer().getExecutorList();

        // 全域当天处理总数，失败总数
        saveOrUpdateDomainProcessCount(
                new ZkStatistics(statisticsModel.getZkClusterDailyCountAnalyzer().getTotalCount(),
                        statisticsModel.getZkClusterDailyCountAnalyzer().getErrorCount()), zkCluster.getZooKeeperAddress());

        // 失败率Top10的域列表
        saveOrUpdateTop10FailDomain(statisticsModel.getDomainStatisticsAnalyzer().getDomainList(),
                zkCluster.getZooKeeperAddress());

        // 稳定性最差的Top10的域列表
        saveOrUpdateTop10UnstableDomain(statisticsModel.getDomainStatisticsAnalyzer().getDomainList(),
                zkCluster.getZooKeeperAddress());

        // 稳定性最差的Top10的executor列表
        saveOrUpdateTop10FailExecutor(executorList, zkCluster.getZooKeeperAddress());

        // 根据失败率Top10的作业列表
        saveOrUpdateTop10FailJob(jobList, zkCluster.getZooKeeperAddress());

        // 最活跃作业Top10的作业列表(即当天执行次数最多的作业)
        saveOrUpdateTop10ActiveJob(jobList, zkCluster.getZooKeeperAddress());

        // 负荷最重的Top10的作业列表
        saveOrUpdateTop10LoadJob(jobList, zkCluster.getZooKeeperAddress());

        // 负荷最重的Top10的Executor列表
        saveOrUpdateTop10LoadExecutor(executorList, zkCluster.getZooKeeperAddress());

        // 异常作业列表 (如下次调度时间已经过了，但是作业没有被调度)
        saveOrUpdateAbnormalJob(statisticsModel.getOutdatedNoRunningJobAnalyzer().getOutdatedNoRunningJobs(),
                zkCluster.getZooKeeperAddress());

        // 超时告警的作业列表
        saveOrUpdateTimeout4AlarmJob(statisticsModel.getTimeout4AlarmJobAnalyzer().getTimeout4AlarmJobList(),
                zkCluster.getZooKeeperAddress());

        // 无法高可用的作业列表
        saveOrUpdateUnableFailoverJob(statisticsModel.getUnableFailoverJobAnalyzer().getUnableFailoverJobList(),
                zkCluster.getZooKeeperAddress());


        // 不同作业等级的作业数量
        saveOrUpdateJobRankDistribution(jobList, zkCluster.getZooKeeperAddress());

        // 容器executor数量
        saveOrUpdateExecutorInDockerCount(statisticsModel.getExecutorInfoAnalyzer().getExeInDocker(),
                zkCluster.getZooKeeperAddress());

        // 物理机executor数量
        saveOrUpdateExecutorNotInDockerCount(statisticsModel.getExecutorInfoAnalyzer().getExeNotInDocker(),
                zkCluster.getZooKeeperAddress());

        // 作业数量
        saveOrUpdateJobCount(jobList.size(), zkCluster.getZooKeeperAddress());
    }

    private void saveOrUpdateJobCount(int jobCount, String zkAddr) {
        try {
            String jobCountString = JSON.toJSONString(jobCount);
            StatisticsDto jobCountFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.JOB_COUNT, zkAddr);
            if (jobCountFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.JOB_COUNT, zkAddr,
                        jobCountString);
                statisticsService.create(ss);
            } else {
                jobCountFromDB.setResult(jobCountString);
                statisticsService.updateByPrimaryKey(jobCountFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateExecutorNotInDockerCount(int exeNotInDocker, String zkAddr) {
        try {
            String exeNotInDockerString = JSON.toJSONString(exeNotInDocker);
            StatisticsDto exeNotInDockerFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.EXECUTOR_NOT_IN_DOCKER_COUNT, zkAddr);
            if (exeNotInDockerFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.EXECUTOR_NOT_IN_DOCKER_COUNT,
                        zkAddr, exeNotInDockerString);
                statisticsService.create(ss);
            } else {
                exeNotInDockerFromDB.setResult(exeNotInDockerString);
                statisticsService.updateByPrimaryKey(exeNotInDockerFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateExecutorInDockerCount(int exeInDocker, String zkAddr) {
        try {
            String exeInDockerString = JSON.toJSONString(exeInDocker);
            StatisticsDto exeInDockerFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.EXECUTOR_IN_DOCKER_COUNT, zkAddr);
            if (exeInDockerFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.EXECUTOR_IN_DOCKER_COUNT, zkAddr,
                        exeInDockerString);
                statisticsService.create(ss);
            } else {
                exeInDockerFromDB.setResult(exeInDockerString);
                statisticsService.updateByPrimaryKey(exeInDockerFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateJobRankDistribution(List<JobStatistics> jobList, String zkBsKey) {
        try {
            Map<Integer, Integer> jobDegreeCountMap = new HashMap<>();
            for (JobStatistics jobStatistics : jobList) {
                int jobDegree = jobStatistics.getJobDegree();
                Integer count = jobDegreeCountMap.get(jobDegree);
                jobDegreeCountMap.put(jobDegree, count == null ? 1 : count + 1);
            }
            String jobDegreeMapString = JSON.toJSONString(jobDegreeCountMap);
            StatisticsDto jobDegreeMapFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.JOB_RANK_DISTRIBUTION, zkBsKey);
            if (jobDegreeMapFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.JOB_RANK_DISTRIBUTION, zkBsKey,
                        jobDegreeMapString);
                statisticsService.create(ss);
            } else {
                jobDegreeMapFromDB.setResult(jobDegreeMapString);
                statisticsService.updateByPrimaryKey(jobDegreeMapFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateTop10FailExecutor(List<ExecutorStatistics> executorList, String zkAddr) {
        try {
            executorList = DashboardServiceHelper.sortExecutorByFailureRate(executorList);
            List<ExecutorStatistics> top10FailExecutor = executorList
                    .subList(0, executorList.size() > 9 ? 10 : executorList.size());
            String top10FailExecutorJsonString = JSON.toJSONString(top10FailExecutor);
            StatisticsDto top10FailExecutorFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_FAIL_EXECUTOR, zkAddr);
            if (top10FailExecutorFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_FAIL_EXECUTOR, zkAddr,
                        top10FailExecutorJsonString);
                statisticsService.create(ss);
            } else {
                top10FailExecutorFromDB.setResult(top10FailExecutorJsonString);
                statisticsService.updateByPrimaryKey(top10FailExecutorFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }

    }

    private void saveOrUpdateTop10FailDomain(List<DomainStatistics> domainList, String zkAddr) {
        try {
            domainList = DashboardServiceHelper.sortDomainByAllTimeFailureRate(domainList);
            List<DomainStatistics> top10FailDomainList = domainList
                    .subList(0, domainList.size() > 9 ? 10 : domainList.size());
            String top10FailDomainJsonString = JSON.toJSONString(top10FailDomainList);
            StatisticsDto top10FailDomainFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_FAIL_DOMAIN, zkAddr);
            if (top10FailDomainFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_FAIL_DOMAIN, zkAddr,
                        top10FailDomainJsonString);
                statisticsService.create(ss);
            } else {
                top10FailDomainFromDB.setResult(top10FailDomainJsonString);
                statisticsService.updateByPrimaryKey(top10FailDomainFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateTop10UnstableDomain(List<DomainStatistics> domainList, String zkAddr) {
        try {
            domainList = DashboardServiceHelper.sortDomainByShardingCount(domainList);
            List<DomainStatistics> top10UnstableDomain = domainList
                    .subList(0, domainList.size() > 9 ? 10 : domainList.size());
            String top10UnstableDomainJsonString = JSON.toJSONString(top10UnstableDomain);
            StatisticsDto top10UnstableDomainFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_UNSTABLE_DOMAIN, zkAddr);
            if (top10UnstableDomainFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_UNSTABLE_DOMAIN, zkAddr,
                        top10UnstableDomainJsonString);
                statisticsService.create(ss);
            } else {
                top10UnstableDomainFromDB.setResult(top10UnstableDomainJsonString);
                statisticsService.updateByPrimaryKey(top10UnstableDomainFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateTop10FailJob(List<JobStatistics> jobList, String zkAddr) {
        try {
            jobList = DashboardServiceHelper.sortJobByAllTimeFailureRate(jobList);
            List<JobStatistics> top10FailJob = jobList.subList(0, jobList.size() > 9 ? 10 : jobList.size());
            String top10FailJobJsonString = JSON.toJSONString(top10FailJob);
            StatisticsDto top10FailJobFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_FAIL_JOB, zkAddr);
            if (top10FailJobFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_FAIL_JOB, zkAddr,
                        top10FailJobJsonString);
                statisticsService.create(ss);
            } else {
                top10FailJobFromDB.setResult(top10FailJobJsonString);
                statisticsService.updateByPrimaryKey(top10FailJobFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateTop10ActiveJob(List<JobStatistics> jobList, String zkAddr) {
        try {
            jobList = DashboardServiceHelper.sortJobByDayProcessCount(jobList);
            List<JobStatistics> top10ActiveJob = jobList.subList(0, jobList.size() > 9 ? 10 : jobList.size());
            String top10ActiveJobJsonString = JSON.toJSONString(top10ActiveJob);
            StatisticsDto top10ActiveJobFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_ACTIVE_JOB, zkAddr);
            if (top10ActiveJobFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_ACTIVE_JOB, zkAddr,
                        top10ActiveJobJsonString);
                statisticsService.create(ss);
            } else {
                top10ActiveJobFromDB.setResult(top10ActiveJobJsonString);
                statisticsService.updateByPrimaryKey(top10ActiveJobFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateTop10LoadJob(List<JobStatistics> jobList, String zkAddr) {
        try {
            jobList = DashboardServiceHelper.sortJobByLoadLevel(jobList);
            List<JobStatistics> top10LoadJob = jobList.subList(0, jobList.size() > 9 ? 10 : jobList.size());
            String top10LoadJobJsonString = JSON.toJSONString(top10LoadJob);
            StatisticsDto top10LoadJobFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_LOAD_JOB, zkAddr);
            if (top10LoadJobFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_LOAD_JOB, zkAddr,
                        top10LoadJobJsonString);
                statisticsService.create(ss);
            } else {
                top10LoadJobFromDB.setResult(top10LoadJobJsonString);
                statisticsService.updateByPrimaryKey(top10LoadJobFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateTop10LoadExecutor(List<ExecutorStatistics> executorList, String zkAddr) {
        try {
            executorList = DashboardServiceHelper.sortExecutorByLoadLevel(executorList);
            List<ExecutorStatistics> top10LoadExecutor = executorList
                    .subList(0, executorList.size() > 9 ? 10 : executorList.size());
            String top10LoadExecutorJsonString = JSON.toJSONString(top10LoadExecutor);
            StatisticsDto top10LoadExecutorFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TOP_10_LOAD_EXECUTOR, zkAddr);
            if (top10LoadExecutorFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TOP_10_LOAD_EXECUTOR, zkAddr,
                        top10LoadExecutorJsonString);
                statisticsService.create(ss);
            } else {
                top10LoadExecutorFromDB.setResult(top10LoadExecutorJsonString);
                statisticsService.updateByPrimaryKey(top10LoadExecutorFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateDomainProcessCount(ZkStatistics zks, String zkAddr) {
        try {
            String domainListJsonString = JSON.toJSONString(zks);
            StatisticsDto domainProcessCountFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.DOMAIN_PROCESS_COUNT_OF_THE_DAY, zkAddr);
            if (domainProcessCountFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.DOMAIN_PROCESS_COUNT_OF_THE_DAY,
                        zkAddr, domainListJsonString);
                statisticsService.create(ss);
            } else {
                domainProcessCountFromDB.setResult(domainListJsonString);
                statisticsService.updateByPrimaryKey(domainProcessCountFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void saveOrUpdateAbnormalJob(List<AbnormalJob> unnormalJobList, String zkAddr) {
        try {
            unnormalJobList = DashboardServiceHelper.sortUnnormaoJobByTimeDesc(unnormalJobList);
            StatisticsDto unnormalJobFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.UNNORMAL_JOB, zkAddr);
            if (unnormalJobFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.UNNORMAL_JOB, zkAddr,
                        JSON.toJSONString(unnormalJobList));
                statisticsService.create(ss);
            } else {
                List<AbnormalJob> oldUnnormalJobList = JSON
                        .parseArray(unnormalJobFromDB.getResult(), AbnormalJob.class);
                // 再次同步数据库中最新的read状态
                dealWithReadStatus(unnormalJobList, oldUnnormalJobList);
                unnormalJobFromDB.setResult(JSON.toJSONString(unnormalJobList));
                statisticsService.updateByPrimaryKey(unnormalJobFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void dealWithReadStatus(List<AbnormalJob> unnormalJobList, List<AbnormalJob> oldUnnormalJobList) {
        if (oldUnnormalJobList == null || oldUnnormalJobList.isEmpty()) {
            return;
        }
        for (AbnormalJob example : unnormalJobList) {
            AbnormalJob equalOld = DashboardServiceHelper.findEqualAbnormalJob(example, oldUnnormalJobList);
            if (equalOld != null) {
                example.setRead(equalOld.isRead());
            }
        }
    }

    private void saveOrUpdateTimeout4AlarmJob(List<Timeout4AlarmJob> timeout4AlarmJobList, String zkAddr) {
        try {
            String timeout4AlarmJobJsonString = JSON.toJSONString(timeout4AlarmJobList);
            StatisticsDto timeout4AlarmJobFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.TIMEOUT_4_ALARM_JOB, zkAddr);
            if (timeout4AlarmJobFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.TIMEOUT_4_ALARM_JOB, zkAddr,
                        timeout4AlarmJobJsonString);
                statisticsService.create(ss);
            } else {
                List<Timeout4AlarmJob> oldTimeout4AlarmJobs = JSON
                        .parseArray(timeout4AlarmJobFromDB.getResult(), Timeout4AlarmJob.class);
                // 再次同步数据库中最新的read状态
                dealWithReadStatus4Timeout4AlarmJob(timeout4AlarmJobList, oldTimeout4AlarmJobs);
                timeout4AlarmJobFromDB.setResult(JSON.toJSONString(timeout4AlarmJobList));
                statisticsService.updateByPrimaryKey(timeout4AlarmJobFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void dealWithReadStatus4Timeout4AlarmJob(List<Timeout4AlarmJob> jobList,
                                                     List<Timeout4AlarmJob> oldJobList) {
        if (oldJobList == null || oldJobList.isEmpty()) {
            return;
        }
        for (Timeout4AlarmJob job : jobList) {
            Timeout4AlarmJob oldJob = DashboardServiceHelper.findEqualTimeout4AlarmJob(job, oldJobList);
            if (oldJob != null) {
                job.setRead(oldJob.isRead());
            }
        }
    }

    private void saveOrUpdateUnableFailoverJob(List<AbnormalJob> unableFailoverJobList, String zkAddr) {
        try {
            String unableFailoverJobJsonString = JSON.toJSONString(unableFailoverJobList);
            StatisticsDto unableFailoverJobFromDB = statisticsService
                    .findStatisticsByNameAndZkList(StatisticsTableKeyConstant.UNABLE_FAILOVER_JOB, zkAddr);
            if (unableFailoverJobFromDB == null) {
                StatisticsDto ss = new StatisticsDto(StatisticsTableKeyConstant.UNABLE_FAILOVER_JOB, zkAddr,
                        unableFailoverJobJsonString);
                statisticsService.create(ss);
            } else {
                unableFailoverJobFromDB.setResult(unableFailoverJobJsonString);
                statisticsService.updateByPrimaryKey(unableFailoverJobFromDB);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }


}
